package articleEvents

import (
	"context"
	"github.com/IBM/sarama"
	"github.com/ecodeclub/ekit/syncx/atomicx"
	"gorm.io/gorm"
	"jk-time/webook/interactive/repository"
	"jk-time/webook/pkg/canalx"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/migrator"
	"jk-time/webook/pkg/migrator/events"
	"jk-time/webook/pkg/migrator/validator"
	"jk-time/webook/pkg/saramax"
	"time"
)

type MysqlBinlogConsumer[T migrator.Entity] struct {
	client   sarama.Client
	repo     repository.InteractiveRepository
	dstFirst *atomicx.Value[bool]
	l        logger.Logger
	srcToDst *validator.CanalIncrValidator[T]
	dstToSrc *validator.CanalIncrValidator[T]
}

func NewMysqlBinlogConsumer[T migrator.Entity](client sarama.Client, repo repository.InteractiveRepository,
	l logger.Logger, src *gorm.DB, dst *gorm.DB, p events.Producer) *MysqlBinlogConsumer[T] {
	return &MysqlBinlogConsumer[T]{
		client:   client,
		repo:     repo,
		l:        l,
		dstFirst: atomicx.NewValueOf(false),
		srcToDst: validator.NewCanalIncrValidator[T](validator.NewBaseValidator(src, dst, "SRC", l, p)),
		dstToSrc: validator.NewCanalIncrValidator[T](validator.NewBaseValidator(dst, src, "DST", l, p)),
	}
}

func (m *MysqlBinlogConsumer[T]) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("inter_binlog", m.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{"example"},
			saramax.NewHandler[canalx.Message[T]]("inter_binlog", m.Consume))
		if err != nil {
			m.l.Error("退出了消费循环,异常", logger.Error(err))
		}
	}()
	return nil
}
func (m *MysqlBinlogConsumer[T]) StartBatch() error {
	return nil
}
func (m *MysqlBinlogConsumer[T]) Consume(msg *sarama.ConsumerMessage, val canalx.Message[T]) error {
	dstFirst := m.dstFirst.Load()
	var v *validator.CanalIncrValidator[T]
	if dstFirst && val.Database == "webook_intr" {
		// 以目标表为准，过来的也恰好是目标表的 binlog
		// 要校验
		v = m.dstToSrc
	} else if !dstFirst && val.Database == "webook" {
		// 以源表为准，过来的也恰好是源表的 binlog
		// 要校验
		v = m.srcToDst
	}
	if v == nil {
		return nil
	}
	for _, data := range val.Data {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		err := v.Validate(ctx, data.ID())
		cancel()
		if err != nil {
			return err
		}
	}
	return nil
}
func (k *MysqlBinlogConsumer[T]) setDstFirst() {
	k.dstFirst.Store(true)
}
